package com.jeffse.light.stream.core.data.operator.Covert;

import com.jeffse.light.stream.core.constans.TransformConstans;
import com.jeffse.light.stream.core.data.descriptor.operatorDescriptor.OperatorDescriptor;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

public final class Convert implements Serializable {
    private final Map<String, BaseConvert> map = new HashMap<>();
    private static Convert INSTANCE = new Convert();

    /**
     * 初始化mapping
     */
    public Convert() {
        // 当类型是A 调用convertA
        map.put(TransformConstans.MAP, new MapConvert());
        map.put(TransformConstans.FLAT_MAP, new FlatMapConvert());
        map.put(TransformConstans.PROCESS, new ProcessConvert());
        map.put(TransformConstans.KEYED_PROCESS, new KeyedProcessConvert());
    }

    public static Convert getINSTANCE() {
        return INSTANCE;
    }

    public void handleData(OperatorDescriptor operatorDescriptor, Object value) throws Exception {
        map.get(operatorDescriptor.getName()).convert(operatorDescriptor, value);
    }
}
